 1.Spark Streaming之与Kafka整合
   
   官网：http://spark.apache.org/docs/2.4.5/streaming-kafka-integration.html
   针对不同的spark、kafka版本，集成处理数据的方式分为两种：Receiver Approach
和Direct Approach，不同集成版本处理方式的支持，可参考下图：
   对Kafka的支持分为两个版本08（在高版本中将被废弃）、010，两个版本不兼容。
 
 2.Kafka-08 接口
   
   1).Receiver based Approach
   基于 Receiver 的方式使用 Kafka 旧版消费者高阶API实现。
   对于所有的 Receiver，通过 Kafka 接收的数据被存储于 Spark 的 Executors上，底
层是写入BlockManager中，默认200ms生成一个block(spark.streaming.blockInterval)
。然后由 Spark Streaming 提交的 job 构建BlockRDD，最终以 Spark Core任务的形式
运行。对应 Receiver方式，有以下几点需要注意：
      Receiver 作为一个常驻线程调度到 Executor上运行，占用一个cpu
      Receiver 个数由KafkaUtils.createStream调用次数决定，一次一个 Receiver
	  kafka中的topic分区并不能关联产生在spark streaming中的rdd分区。增加在
KafkaUtils.createStream()中的指定的topic分区数，仅仅增加了单个receiver消费
的topic的线程数，它不会增加处理数据中的并行的spark的数量。【 即：topicMap
[topic,num_threads]中，value对应的数值是每个topic对应的消费线程数】
      receiver默认200ms生成一个block，可根据数据量大小调整block生成周期。一
个block对应RDD一个分区。
      receiver接收的数据会放入到BlockManager，每个 Executor 都会有一个
BlockManager实例，由于数据本地性，那些存在 Receiver 的 Executor 会被调
度执行更多的 Task，就会导致某些executor比较空闲
      默认情况下，Receiver是可能丢失数据的。可以通过设置spark.streaming.receiver
.writeAheadLog.enable为true开启预写日志机制，将数据先写入一个可靠地分布式
文件系统(如HDFS)，确保数据不丢失，但会损失一定性能  
   Kafka-08 接口(Receiver方式) <=Spark1.1：
      Offset保存在ZK中，系统管理
      对应Kafka的版本 0.8.2.1+
      接口底层实现使用 Kafka 旧版消费者高阶API
      DStream底层实现为BlockRDD
   Kafka-08 接口(Receiver with WAL)Spark1.2：
      增强了故障恢复的能力
      接收的数据与Dirver的元数据保存到HDFS
      增加了流式应用处理的延迟
   2).Direct Approach
   Direct Approach是 Spark Streaming不使用Receiver集成kafka的方式，在企业生产
环境中使用较多。相较于Receiver，有以下特点：
      不使用 Receiver。减少不必要的CPU占用；减少了 Receiver接收数据写入BlockManager
，然后运行时再通过blockId、网络传输、磁盘读取等来获取数据的整个过程，提升了效率
；无需WAL，进一步减少磁盘IO；  
      Direct方式生的RDD是KafkaRDD，它的分区数与 Kafka 分区数保持一致，便于把控
并行度 注意：在 Shuffle 或 Repartition 操作后生成的RDD，这种对应关系会失效
      可以手动维护offset，实现 Exactly Once 语义
 
 3.Kafka-010 接口
   
   Spark Streaming与kafka 0.10的整合，和0.8版本的 Direct 方式很像。Kafka的分区
和Spark的RDD分区是一一对应的，可以获取 offsets 和元数据，API 使用起来没有显著
的区别。
   添加依赖：
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
    <version>${spark.version}</version>
</dependency>
   不要手动添加 org.apache.kafka 相关的依赖，如kafka-clients。spark-streaming-
kafka-0-10 已经包含相关的依赖了，不同的版本会有不同程度的不兼容。
   使用kafka010接口从 Kafka 中获取数据：
       Kafka集群
       kafka生产者发送数据
       Spark Streaming程序接收数   
package cn.lagou.Streaming.kafka

import java.util.Properties
import org.apache.kafka.clients.producer.{KafkaProducer,
  ProducerConfig,
  ProducerRecord
}
import org.apache.kafka.common.serialization.StringSerializer

object KafkaProducer {
  def main(args: Array[String]): Unit = {
    // 定义 kafka 参数
    val brokers = "linux121:9092,linux122:9092,linux123:9092"
    val topic = "topicB"
    val prop = new Properties()
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
      classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
      classOf[StringSerializer])
    // KafkaProducer
    val producer = new KafkaProducer[String, String](prop)
    for (i <- 1 to 1000000) {
      val msg = new ProducerRecord[String, String](topic,
        i.toString, i.toString)
      // 发送消息
      producer.send(msg)
      println(s"i = $i")
      Thread.sleep(100)
    }
    producer.close()
  }
}


package cn.lagou.Streaming.kafka

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, 
  KafkaUtils, LocationStrategies}

object KafkaDStream1 {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    // 定义kafka相关参数
    val groupId: String = "mygroup01"
    val topics: Array[String] = Array("topicB")
    val kafkaParams: Map[String, Object] = getKafkaConsumerParams(groupId)
    // 创建DStream
    val dstream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )

    // DStream转换&输出
    dstream.foreachRDD{(rdd, time) =>
      println(s"************ rdd.count = ${rdd.count()}; " +
        s"time =$time*********")
    }


    // 启动作业
    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaConsumerParams(groupid: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux121:9092,linux122:9092,linux123:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupid,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean),
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
    )
  }
}
   
   LocationStrategies(本地策略)
      LocationStrategies.PreferBrokers：如果 Executor 在 kafka 集群中的某些节点
上，可以使用这种策略。此时Executor 中的数据会来自当前broker节点
      LocationStrategies.PreferConsistent：大多数情况下使用的策略，将Kafka分
区均匀的分布在Spark集群的 Executor上
      LocationStrategies.PreferFixed：如果节点之间的分区有明显的分布不均，使
用这种策略。通过一个map指定将 topic 分区分布在哪些节点中
   ConsumerStrategies(消费策略)
      ConsumerStrategies.Subscribe，用来订阅一组固定topic
      ConsumerStrategies.SubscribePattern，使用正则来指定感兴趣的topic
      ConsumerStrategies.Assign，指定固定分区的集合
   这三种策略都有重载构造函数，允许指定特定分区的起始偏移量；使用 Subscribe
或 SubscribePattern 在运行时能实现分区自动发现。
   Kafka相关命令：
   # 创建 topic
kafka-topics.sh --zookeeper
linux121:2181,linux122:2181,linux123:2181 \
--create --topic topicB --replication-factor 2 --partitions 3
   
   # 显示 topic 信息
kafka-topics.sh --zookeeper
linux121:2181,linux122:2181,linux123:2181 --topic topicA --describe
   
   # 检查 topic 的最大offset
   kafka-run-class.sh kafka.tools.GetOffsetShell \
--broker-list linux121:9092,linux122:9092,linux123:9092 --topic
topicA --time -1
   
   # 列出所有的消费者
kafka-consumer-groups.sh --bootstrap-server linux121:9092  --list
   
   # 检查消费者的offset
kafka-consumer-groups.sh --bootstrap-server
linux121:9092,linux122:9092,linux123:9092 \
--describe --group mygroup01
   
   # 重置消费者offset
kafka-consumer-groups.sh --bootstrap-server
linux121:9092,linux122:9092,linux123:9092 \
--group group01 --reset-offsets --execute --to-offset 0 --topic
topicA
   
   